package com.mrogrape.messager;

import java.io.BufferedWriter;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

/**
 * MessageService使用UDP协议通过网络传送数据并监听来自网络的UDP数据报。
 * 
 * <h3>工作原理</h3>
 * <p>
 * MessageService使用不同线程来分别处理数据报的发送、接收、分派。
 * 而对这些操作中产生的事件也通过一个独立的事件分派线程来进行指派。这些线程间通过同步队列来协调。
 * 这种方案可以最大限度的减小不同任务间的相互影响。避免出现某个过程因为某些原因处于
 * 等待而影响其他的操作。
 * 
 * <h3>消息和消息监听器</h3>
 * <p>
 * 因为UDP数据报的特性，通过MessagerService发送和接收的数据的最小单位被称为“消息”。
 * 消息由接口{@link com.mrogrape.messager.MessageBundle}声明。每个消息对象拥有一个
 * 唯一的，不可变的ID。在内部以“键-值”的方式储存数据。用户不可直接创建消息的对象。
 * 应通过具体的实现或者使用MessagerService.createMessageBundle()来创建一个消息对象。
 * <p>
 * 如果希望接收消息，那么可以实现MessageListener接口并添加。MessageListener中的方法
 * 会被消息分派线程调用。所以请注意不要在其中实现需要较长时间处理的操作。否则待指派的
 * 数据包数目达到消息分派队列的上限后，后续的数据包将被丢弃。
 * 
 * <h3>事件</h3>
 * MessageService服务的启动，停止，状态更改和消息的发送，指派都会产生事件。如果对这些事件
 * 感兴趣可以实现ServiceListener（服务事件）或更具体的MessageStateListener(消息事件）。
 * 同理，事件指派线程中也不应该存在需要长时间操作的行为以免阻塞线程而对后面的事件处理产生影响。
 * 
 * @author 
 *
 */
public class MessagerService {
	
	/** 最大消息长度 */
	private static final int MESSAGE_MAX_LENGTH = 32000;
	/** 发送队列默认容量 */
	private static final int SEND_QUEUE_CAPACITY = 200;
	/** 接收队列默认容量 */
	private static final int RECEIVE_QUEUE_CAPACITY = 200;
	/** 消息指派队列默认容量 */
	private static final int DISPATHER_QUEUE_CAPACITY = 200;
	/** 默认情况下，消息发送队列查看一次取消请求所使用的最小时间 */
	private static final int SENDER_CHECKE_TIMEOUT = 1000;
	/** 默认情况下，消息指派队列查看一次取消请求所使用的最小时间 */
	private static final int MSG_DISPATCHER_CHECKE_TIMEOUT = 1000;
	/** 默认情况下，消息指派队列查看一次取消请求所使用的最小时间 */
	private static final int MSG_RECEPTOR_CHECKE_TIMEOUT = 1000;
	
	public MessagerService(int port) throws SocketException, UnknownHostException{
		
		this.udpServerPort = port;
		this.socket = new DatagramSocket(port, InetAddress.getLocalHost());
		this.socket.setSoTimeout(MessagerService.MSG_RECEPTOR_CHECKE_TIMEOUT);
		
		this.msgSndQueue = new ArrayBlockingQueue<MessagePackage>(MessagerService.SEND_QUEUE_CAPACITY);
		this.msgRcvQueue = new ArrayBlockingQueue<MessagePackage>(MessagerService.RECEIVE_QUEUE_CAPACITY);
		this.dispatherQueue = new ArrayBlockingQueue<ServiceEvent>(MessagerService.DISPATHER_QUEUE_CAPACITY);
	}
	
	/**
	 * 创建一个消息包装器。
	 * @return
	 */
	public static MessageBundle createMessageBundle(){
		
		return new Message();
	}
	
	/**
	 * 创建一个消息包装器,并初始化数据。
	 * 
	 * @param data
	 * @return
	 * @throws IOException
	 */
	public static MessageBundle createMessageBundle(byte[] data) throws IOException{
		
		Message msg = new Message();
		msg.decode(data);
		
		return msg;
	}
	
	/**
	 * 开始服务，如果服务已经启动，则不执行任何操作。
	 */
	public synchronized void startService(){
		
			// Check if service is started.
			if (this.serviceStarted) {

				throw new IllegalStateException("The service is already started or it's closed.");
			}

			this.transmitter = new Transmitter(this.socket, this.msgSndQueue);
			this.transmitter.start();

			this.receptor = new Receptor(this, this.socket, this.msgRcvQueue);
			this.receptor.start();

			this.messageDispatcher = new MessageDispatcher(this, this.msgListeners, this.msgRcvQueue);
			this.messageDispatcher.start();

			this.serviceStarted = true;
		
	}
	
	/**
	 * 停止服务。
	 * 
	 * <p>
	 * 此方法调用后服务不会立即停止，它们会在处理完当前周期的任务后才着手退出。
	 */
	public synchronized void stopService(){
		if(!this.serviceStarted){
			
			throw new IllegalStateException("Service has not started.");
		}
		
		this.transmitter.requestCancel();
		this.receptor.requestCancel();
		this.receptor.requestCancel();
		this.messageDispatcher.requestCancel();
		
	}
	
	/**
	 * 立即停止服务，如果服务未启动则抛出IllegalStateException异常。
	 * 
	 * <p>
	 * 注意，这个方法会立即停止服务，丢弃所有队列中的所有数据包。
	 */
	public synchronized void stopServiceImmediately(){
			
			if(!this.serviceStarted){
				
				throw new IllegalStateException("Service has not started.");
			}
			
			this.transmitter.interrupt();
			this.receptor.interrupt();
			this.socket.close();
			this.messageDispatcher.interrupt();
	}
	
	/**
	 * 发送消息。
	 * 
	 * @param msg
	 * @param device
	 * @return
	 */
	public boolean sendMessage(MessageBundle msg, TerminalDevice device){
		
		MessagePackage pkg = new MessagePackage();
		pkg.setMessage(msg);
		pkg.setRecipient(device);
		
		return this.msgSndQueue.offer(pkg);
	}
	
	public void addMessageListener(MessageListener ml){
		
		this.msgListeners.add(ml);
	}
	
	private TerminalDevice createLocalDevice() throws UnknownHostException{
		
		TerminalDevice device = new TerminalDevice();
		device.setAddress(InetAddress.getLocalHost());
		device.setUdpServicePort(this.udpServerPort);
		
		return device;
	}
	
	
	/** UDP协议使用的端口 */
	private int udpServerPort;
	/** 套接字 */
	private DatagramSocket socket;
	
	private boolean serviceStarted;
	
	/** 消息发送器 */
	private Transmitter transmitter;
	/** 消息接收处理器 */
	private Receptor receptor;
	/** 消息调度处理器 */
	private MessageDispatcher messageDispatcher;
	/** 事件调度处理器 */
	private EventDispatcher eventDispatcher;
	
	/** 消息发送队列 */
	private BlockingQueue<MessagePackage> msgSndQueue;
	/** 消息处理队列 */
	private BlockingQueue<MessagePackage> msgRcvQueue;
	/** 事件指派队列 */
	private BlockingQueue<ServiceEvent> dispatherQueue;
	
	/** 消息监听器 */
	private List<MessageListener> msgListeners = Collections.synchronizedList(new LinkedList<MessageListener>());
	/** 服务监听器 */
	private List<ServiceListener> serverListeners = Collections.synchronizedList(new LinkedList<ServiceListener>());
	
	/** 默认的消息实现。 */
	private static class Message implements MessageBundle{
		
		/** serialVersionUID */
		private static final long serialVersionUID = 1L;
		
		public Message(){
			this.id = IDGenerator.generateRandomLongId();
			this.dataset = new Properties();
		}

		@Override
		public long getId() {
			return this.id;
		}

		@Override
		public void putByte(String key, byte value) {
			this.dataset.put(key, String.valueOf(value));
			
		}

		@Override
		public void putInt(String key, int value) {
			this.dataset.put(key, String.valueOf(value));
			
		}

		@Override
		public void putShort(String key, short value) {
			this.dataset.put(key, String.valueOf(value));
			
		}

		@Override
		public void putLong(String key, long value) {
			this.dataset.put(key, String.valueOf(value));
			
		}

		@Override
		public void putChar(String key, char value) {
			this.dataset.put(key, String.valueOf(value));
			
		}

		@Override
		public void putBoolean(String key, boolean value) {
			this.dataset.put(key, String.valueOf(value));
			
		}

		@Override
		public void putDouble(String key, double value) {
			this.dataset.put(key, String.valueOf(value));
			
		}

		@Override
		public void putFloat(String key, float value) {
			this.dataset.put(key, String.valueOf(value));
			
		}

		@Override
		public void putString(String key, String value) {
			this.dataset.put(key, String.valueOf(value));
			
		}

		@Override
		public Byte getByte(String key) {
			
			Object value = this.dataset.get(key);
			return value == null?null:Byte.valueOf(value.toString());
		}

		@Override
		public Integer getInt(String key) {
			Object value = this.dataset.get(key);
			return value == null?null:Integer.valueOf(value.toString());
		}

		@Override
		public Short getShort(String key) {
			Object value = this.dataset.get(key);
			return value == null?null:Short.valueOf(value.toString());
		}

		@Override
		public Long getLong(String key) {
			Object value = this.dataset.get(key);
			return value == null?null:Long.valueOf(value.toString());
		}

		@Override
		public Character getChar(String key) {
			Object value = this.dataset.get(key);
			return value == null?null:Character.valueOf(value.toString().charAt(0));
		}

		@Override
		public Boolean getBoolean(String key) {
			Object value = this.dataset.get(key);
			return value == null?null:Boolean.valueOf(value.toString());
		}

		@Override
		public Double getDouble(String key) {
			Object value = this.dataset.get(key);
			return value == null?null:Double.valueOf(value.toString());
		}

		@Override
		public Float getFloat(String key) {
			Object value = this.dataset.get(key);
			return value == null?null:Float.valueOf(value.toString());
		}

		@Override
		public String getString(String key) {
			Object value = this.dataset.get(key);
			return value == null?null:String.valueOf(value.toString());
		}

		@Override
		public byte[] encode() throws IOException{
			
			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(baos));
			try{
				this.dataset.store(bw, null);
			}finally{
				bw.close();
				baos.close();
			}
			
			return baos.toByteArray();
		}

		@Override
		public void decode(byte[] data) throws IOException {
			this.dataset.clear();
			InputStreamReader isr = new InputStreamReader(new ByteArrayInputStream(data));
			try{
				this.dataset.load(isr);
			}finally{
				isr.close();
			}
		}

		@Override
		public String[] getKeys() {
			// TODO Auto-generated method stub
			return null;
		}
		
		private final long id;
		private Properties dataset;
	}
	
	/** 数据包 */
	private static class MessagePackage{
		
		public TerminalDevice getSender() {
			return this.sender;
		}
		public void setSender(TerminalDevice sender) {
			this.sender = sender;
		}
		public TerminalDevice getRecipient() {
			return this.recipient;
		}
		public void setRecipient(TerminalDevice recipient) {
			this.recipient = recipient;
		}
		public MessageBundle getMessage() {
			return this.message;
		}
		public void setMessage(MessageBundle message) {
			this.message = message;
		}
		/** 发送者 */
		private TerminalDevice sender;
		/** 接收者 */
		private TerminalDevice recipient;
		/** 消息 */
		private MessageBundle message;
	}
	
	/** 消息发送处理器 */
	private static class Transmitter extends Thread{
		
		
		public Transmitter(DatagramSocket socket, BlockingQueue<MessagePackage> msgSndQueue){
			super("Transmitter-"+IDGenerator.generateRandomLongId());
			this.socket = socket;
			this.msgSndQueue = msgSndQueue;
		}

		
		@Override
		public void run() {
			
			while(!this.cancelRequested){
				
				try {
					MessagePackage msgPackage = this.msgSndQueue.poll(MessagerService.SENDER_CHECKE_TIMEOUT, TimeUnit.MILLISECONDS);
					
					if(msgPackage != null){
						this.handleMessagePackage(msgPackage);
					}
				} catch (InterruptedException e) {
					e.printStackTrace();
					break;
				} catch (IOException e) {
					e.printStackTrace();
					break;
				}
			}
			
			Logger.getLogger(Logger.GLOBAL_LOGGER_NAME).info("Transmitter out.");
			
		}
		
		public void requestCancel() {
			this.cancelRequested = true;
		}
		
		public boolean isCancelRequested() {
			return this.cancelRequested;
		}

		private void handleMessagePackage(MessagePackage msgPkg) throws IOException{
			
			// 获取数据
			byte[] data = msgPkg.getMessage().encode();
			
			// 制作UDP数据包
			DatagramPacket dp = new DatagramPacket(data, data.length, msgPkg.getRecipient().getAddress(), msgPkg.getRecipient().getUdpServicePort());
			
			// 发送数据
			this.socket.send(dp);
		}
		
		/** 取消服务请求 */
		private boolean cancelRequested = false;
		
		/** 套接字 */
		private DatagramSocket socket;
		/** 消息发送队列 */
		private BlockingQueue<MessagePackage> msgSndQueue;
	}
	
	/** 消息接收处理器 */
	private static class Receptor extends Thread{
		
		
		public Receptor(MessagerService service, DatagramSocket socket, BlockingQueue<MessagePackage> msgRcvQueue) {
			super("Receptor-"+IDGenerator.generateRandomLongId());
			this.service = service;
			this.socket = socket;
			this.msgRcvQueue = msgRcvQueue;
		}
		
		public void requestCancel() {
			this.cancelRequested = true;
		}
		
		public boolean isCancelRequested() {
			return this.cancelRequested;
		}
		
		@Override
		public void run() {
			
			DatagramPacket dp = null;
			while(!this.cancelRequested){
				
				dp = new DatagramPacket(new byte[MessagerService.MESSAGE_MAX_LENGTH], MessagerService.MESSAGE_MAX_LENGTH);
				try {
					
					this.socket.receive(dp);

					TerminalDevice device = new TerminalDevice();
					device.setAddress(dp.getAddress());
					device.setUdpServicePort(this.service.udpServerPort);
					
					MessagePackage pkg = new MessagePackage();
					pkg.setMessage(MessagerService.createMessageBundle(dp.getData()));
					pkg.setRecipient(this.service.createLocalDevice());
					pkg.setSender(device);
					
					this.msgRcvQueue.offer(pkg);
				}
				// 等待时间超时
				catch(SocketTimeoutException e){
					// DO NOTHING
				}
				catch (SocketException e){
					Logger.getLogger(Logger.GLOBAL_LOGGER_NAME).warning("Receptor has an exception, message: "+e.getMessage());
					return;
				}
				catch (IOException e) {
					Logger.getLogger(Logger.GLOBAL_LOGGER_NAME).warning("Receptor has an exception, message: "+e.getMessage());
					return;
				}
				
			}
			
			Logger.getLogger(Logger.GLOBAL_LOGGER_NAME).info("Receptor out.");
		}
		
		/** 取消服务请求 */
		private boolean cancelRequested = false;
		
		private MessagerService service;
		/** 套接字 */
		private DatagramSocket socket;
		/** 消息处理队列 */
		private BlockingQueue<MessagePackage> msgRcvQueue;
	}
	
	/** 消息调度处理器 */
	private static class MessageDispatcher extends Thread{
		
		
		public MessageDispatcher(MessagerService service, List<MessageListener> msgListeners, BlockingQueue<MessagePackage> msgRcvQueue) {
			super("MessageDispatcher-"+IDGenerator.generateRandomLongId());
			this.service = service;
			this.msgListeners = msgListeners;
			this.msgRcvQueue = msgRcvQueue;
		}

		public void requestCancel() {
			this.cancelRequested = true;
		}
		
		public boolean isCancelRequested() {
			return this.cancelRequested;
		}
		
		@Override
		public void run() {
			
			while(!this.cancelRequested){
				
				try {
					MessagePackage pkg = this.msgRcvQueue.poll(MessagerService.MSG_DISPATCHER_CHECKE_TIMEOUT, TimeUnit.MILLISECONDS);
					
					if(pkg != null){
						
						MessageEvent evt = new MessageEvent();
						
						synchronized(this.msgListeners){
							for(MessageListener l : this.msgListeners){
								evt.setMessage(pkg.getMessage());
								evt.setSender(pkg.getSender());
								evt.setService(this.service);
								l.messageReceived(evt);
							}
						}
					}
					
				} catch (InterruptedException e) {
					e.printStackTrace();
					break;
				}
				
			}
			
			Logger.getLogger(Logger.GLOBAL_LOGGER_NAME).info("MessageDispatcher out.");
		}

		/** 取消服务请求 */
		private boolean cancelRequested = false;
		
		private MessagerService service;
		/** 消息监听器 */
		private List<MessageListener> msgListeners;
		/** 消息处理队列 */
		private BlockingQueue<MessagePackage> msgRcvQueue;
	}
	
	/** 事件调度处理器 */
	private static class EventDispatcher extends Thread{}
	
	
	public static void main(String args[]) throws SocketException, UnknownHostException{
		
		final MessagerService service = new MessagerService(11260);
		
		service.addMessageListener(new MessageListener(){

			@Override
			public void messageReceived(MessageEvent evt) {
				
				try {
					service.stopService();
					System.out.println("Message Received:");
					System.out.println(new String(evt.getMessage().encode()).trim());
					System.out.println("EOF");
					
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				
			}});
		
		service.startService();
		
		MessageBundle msg = MessagerService.createMessageBundle();
		
		msg.putString("string", "String Text");
		
		TerminalDevice device = service.createLocalDevice();
		
		service.sendMessage(msg, device);
		
		
		
		System.out.println("Main terminated");
		
	}
	
}
